package com.edu.yx.app.App_03_DwdApp.Dwd_03_Trade;

import com.edu.yx.app.APP_01_BaseApp.BaseSqlApp;
import com.edu.yx.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static com.edu.yx.common.Constant.*;

public class Dwd_01_Trade_OrderInfoDetail extends BaseSqlApp {

    public static void main(String[] args) {
        new Dwd_01_Trade_OrderInfoDetail().init(
                3031,
                2,
                "Dwd_01_Trade_OrderInfoDetail"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        //1.ods拿数据，只要insert
        readOdsDb(tEnv,"Dwd_01_Trade_OrderInfoDetail");

        //读取课程信息
        readCourseInfo(tEnv);

        //读取学科信息
        readSubjectInfo(tEnv);

        //读取kafka的页面信息，获取order页面的来源
        tEnv.executeSql("create table page(" +
                " `common` map<string, string>, " +
                " `page` map<string, string>, " +
                "`ts` bigint" +
                ")" + SQLUtil.getKafkaSource(TOPIC_DWD_TRAFFIC_PAGE, "Dwd_01_Trade_OrderInfoDetail")
        );

        Table source = tEnv.sqlQuery("select " +
                "page['item'] order_id, " +
                "common['sc'] source " +
                " from page " +
                " where `page`['page_id']='order' " +
                " and `page`['item_type']='order_id' "
        );

        tEnv.createTemporaryView("source",source);


        //2.补充维度信息：学科，类别、课程（order_detail有）、来源（kafka的dwd流量域的page上有）
        //将order_detail的信息补充完整
        Table orderDetail = tEnv.sqlQuery("select " +
                " cast(data['id'] as string) order_detail_id," +
                " cast(data['order_id'] as string) order_id," +
                " cast(data['user_id'] as string) user_id," +
                " cast(data['course_id'] as string) course_id," +
                " cast(ci.subject_id as string) subject_id," +
                " cast(si.category_id as string) category_id," +
                " sc.source," +
                " od.ts, " +
                " od.pt " +

                " from ods_db od " +
                " join course_info for system_time as of od.pt as ci on cast(od.data['course_id'] as bigint) = ci.id " +
                " join subject_info for system_time as of od.pt as si on ci.subject_id = si.id " +
                " join source sc on sc.order_id = od.data['order_id'] " +

                " where `database` = 'edu' " +
                " and `table` = 'order_detail' " +
                " and `type` = 'insert' "
        );
        tEnv.createTemporaryView("order_detail",orderDetail);


        //3.读取ods的order_info信息 补全省份信息
        Table orderInfo = tEnv.sqlQuery("select " +
                "data['id'] order_id," +
                "data['province_id'] province_id," +
                "cast(data['final_amount'] as bigint) final_amount," +
                "ts " +
                " from ods_db  " +
                " where `database` = 'edu' " +
                " and `table` = 'order_info' " +
                " and `type` = 'insert' "
        );
        tEnv.createTemporaryView("order_info",orderInfo);

        //4. order_info join order_detail
        Table result = tEnv.sqlQuery("select " +
                " oi.order_id," +
                " od.user_id," +
                " od.course_id," +
                " od.subject_id," +
                " od.category_id," +
                " od.source," +
                " oi.province_id," +
                " oi.final_amount," +
                " oi.ts" +
                " from order_info oi " +
                " join order_detail od on cast(oi.order_id as string) = od.order_id "

        );

        //5.创建表写入kafka
        tEnv.executeSql("create table dwd_trade_order_info(" +
                "order_id string," +
                "user_id string," +
                "course_id string," +
                "subject_id string," +
                "category_id string," +
                "source string," +
                "province_id string," +
                "final_amount bigint," +
                "ts bigint" +
                ")" + SQLUtil.getKafkaSink(TOPIC_DWD_TRADE_ORDER_INFO)
        );

        result.executeInsert("dwd_trade_order_info");



    }
}
